#include "stdafx.h"

namespace Melting
{
	namespace Parallel
	{ 
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		WorkerThread::WorkerThread( UINT32 _iCapacity )
		:	Thread			()
		,	miRunningMode	( RUN_ONCE )
		,	miNbWorks		( 0 )	

	#ifdef WORKER_THREAD_IMPL_LOCKFREE
		,	mWorkQueue		( _iCapacity )

	#else
		,	miNbWorksReady	( 0 )
		,	mMutex			()
		,	mlpWorks		()
	#endif
		,	mAbortRequested( 0 )
		{
		}
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		WorkerThread::~WorkerThread( void )
		{
			if( RUN_FOREVER == miRunningMode.Get() )
			{
				mAbortRequested.Increment();
				Thread::Wait();
			}
		}
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		bool WorkerThread::Run( RunningMode _eMode )
		{ 
			miRunningMode.Set( (INT32) _eMode );

			return Thread::Run( & WorkerThread::Execute, this );
		}
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		void WorkerThread::AddWork( Work* _pWork )
		{
			miNbWorks.Increment();

	#ifdef WORKER_THREAD_IMPL_LOCKFREE

			mWorkQueue.Enqueue( _pWork );
	#else
			mMutex.Lock();
			{ 
				mlpWorks.push_back( _pWork );
			}
			mMutex.Unlock();

			miNbWorksReady.Increment();
	#endif
		} 
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		INT32 WorkerThread::GetNbWorks( void ) const
		{
			return miNbWorks.Get();
		} 
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		WorkerThread::RunningMode	WorkerThread::GetRunningMode( void ) const
		{
			return (RunningMode) miRunningMode.Get();
		} 
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		//									WORKER THREAD SPECIFIC

		INT32 WorkerThread::GetNbWorksReady( void ) const
		{
	#ifdef WORKER_THREAD_IMPL_LOCKFREE

			return mWorkQueue.GetSize();

	#else
			return miNbWorksReady.Get();

	#endif
		}

		void WorkerThread::ProcessNextWork( void )
		{
	#ifdef WORKER_THREAD_IMPL_LOCKFREE
			Work* pWork = mWorkQueue.Dequeue();
			
	#else
			Work* pWork = mlpWorks.front();
			mMutex.Lock();
			{ 
				mlpWorks.pop_front();
			}
			mMutex.Unlock();
			miNbWorksReady.Decrement();

	#endif
	 
	//        printf(" %p Processes : \n", (void*) this->GetHandle() );
			pWork->Execute();

			pWork->OnExecuted();
			//pWork->~Work();

			miNbWorks.Decrement();

	//        delete pWork;
		}

		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		void WorkerThread::ProcessAllWorks( void )
		{
			while( GetNbWorksReady() > 0 ) 
			{
				ProcessNextWork();
			}
		}
		//-------------------------------------------------------------------------------------------------
		//-------------------------------------------------------------------------------------------------
		ThreadFunctionReturn WorkerThread::Execute( void* _pWorker )
		{
			WorkerThread* pWorker = static_cast< WorkerThread* > ( _pWorker );

			if( WorkerThread::RUN_ONCE == pWorker->GetRunningMode() )
			{
				pWorker->ProcessAllWorks();
			}
			else
			{
				// vars used to minimize the access to the atomic used for abortion
				const int	LOOP_COUNT_BEFORE_ABORTION_CHECK = 100000;
				int			iLoopCnt = 0;
	 
				// check for abortion request
				while( 0 == pWorker->mAbortRequested.Get() )
				{	
					// keep on working !
					while( iLoopCnt < LOOP_COUNT_BEFORE_ABORTION_CHECK )
					{
						while( pWorker->GetNbWorksReady() > 0 ) 
						{
							pWorker->ProcessNextWork();
						}

						// NDAL : dirty ?
						//SwitchThread();
						Parallel::YieldProcessor();
								 
						iLoopCnt++;
					}

					// reset 
					iLoopCnt = 0;
				}
			}

			return (ThreadFunctionReturn)0;
		}
	}// namespace Parallel
}// namespace Melting